package org.smartboot.flow.manager.trace;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.smartboot.flow.core.ExecutionListenerRegistry;
import org.smartboot.flow.core.manager.EngineManager;
import org.smartboot.flow.core.util.AssertUtil;
import org.smartboot.flow.manager.FlatEngine;
import org.smartboot.flow.manager.FlatManager;
import org.smartboot.flow.manager.HostUtils;
import org.smartboot.flow.manager.ManagerConfiguration;
import org.smartboot.flow.manager.ManagerConstants;
import org.smartboot.flow.manager.NamedThreadFactory;
import org.smartboot.flow.manager.UpdateContentTask;
import org.smartboot.http.client.HttpClient;
import org.smartboot.http.client.HttpPost;
import org.smartboot.http.common.enums.HeaderNameEnum;

import java.io.ByteArrayOutputStream;
import java.io.PrintWriter;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author qinluo
 * @date 2023-02-07
 * @since 1.0.7
 */
public class TraceReporter {

    private static final Logger LOGGER = LoggerFactory.getLogger(TraceReporter.class);
    private final ScheduledThreadPoolExecutor executorService = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("export-trace"));

    /**
     * Report idle in mills.
     */
    protected long idle = 5000L;

    /**
     * 服务端地址
     */
    private String serverAddress;

    /**
     * 超时时间
     */
    private long timeout;
    private String host;
    private int port;
    private Double radio;
    private TraceSampleStrategy sampleStrategy;
    private ManagerExecutionListener executionListener;

    private Map<String, String> headers;
    private final BlockingQueue<TraceData> traceQueue = new ArrayBlockingQueue<>(20000);

    public long getIdle() {
        return idle;
    }

    public void setIdle(long idle) {
        AssertUtil.isTrue(idle > 0, "idle must great than zero");
        this.idle = idle;
    }

    public void export() {
        try {
            this.doExport();
        } catch (Throwable ex){
            LOGGER.error("report trace failed", ex);
        }
    }

    public void doExport() {
        List<TraceData> traces = new ArrayList<>();
        if (traceQueue.drainTo(traces) < 0) {
            return;
        }

        EngineManager defaultManager = EngineManager.defaultManager();

        for (TraceData trace : traces) {
            HttpClient httpClient = new HttpClient(host, port);
            httpClient.configuration().connectTimeout((int) timeout);
            HttpPost post = httpClient.post(ManagerConstants.REPORT_TRACE);

            if (headers != null) {
                headers.forEach((key, value) -> post.header().add(key, value));
            }

            TraceReportRequest request = new TraceReportRequest();
            request.setAddress(HostUtils.getHostIp());
            request.setHost(HostUtils.getHostName());
            request.setTimestamp(System.currentTimeMillis());
            FlatEngine flatEngine = FlatManager.getInstance().getFlatEngine(defaultManager.getEngineModel(trace.getEngineName()).getSource());
            request.setMd5(flatEngine.getMd5());
            request.setTraceId(trace.getTraceId());
            request.setEngineName(trace.getEngineName());
            request.setSuccess(trace.getEx() == null);
            request.setTraceTime(trace.getTraceTime());
            request.setEndTime(trace.getEndTime());
            if (!request.getSuccess()) {
                request.setEx(serialExToString(trace.getEx()));
            }
            request.setRequest(JSON.toJSONString(trace.getRequest()));
            request.setResult(JSON.toJSONString(trace.getResult()));

            if (flatEngine.getReportContent()) {
                request.setContent(flatEngine.getContent());
            }

            JSONArray ja = new JSONArray();

            trace.getComponents().forEach((k, v) -> {
                JSONObject item = new JSONObject();
                item.put("name", k.getName());
                item.put("type", k.getType());
                item.put("start", v.getStart());
                item.put("escape", v.getEscape());
                item.put("ex", serialExToString(v.getEx()));
                item.put("rollbackStart", v.getRollbackStart());
                item.put("rollbackEnd", v.getRollbackEnd());

                ja.add(item);
            });

            request.setJson(ja);

            String json = JSON.toJSONString(request, SerializerFeature.WriteEnumUsingToString);
            byte[] bytes = json.getBytes(StandardCharsets.UTF_8);

            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("report trace data, engine = {}, data = {}", trace.getEngineName(), JSON.toJSONString(request));
            }

            post.header().add(HeaderNameEnum.CONTENT_TYPE.getName(), "application/json;charset=UTF-8").add(HeaderNameEnum.CONTENT_LENGTH.getName(), String.valueOf(bytes.length));

            // Use body stream write.
            post.body().write(bytes, 0, bytes.length).done().onSuccess(httpResponse -> {
                        LOGGER.info("send trace success");
                    })
                    .onFailure(throwable -> {
                        LOGGER.error("send trace failed", throwable);
                    });

        }
    }

    private String serialExToString(Throwable ex) {
        if (ex == null) {
            return null;
        }

        int maxDepth = ManagerConfiguration.reportMaxStackDepth;
        StackTraceElement[] stackTrace = ex.getStackTrace();
        if (stackTrace.length > maxDepth) {
            StackTraceElement[] newTrace = new StackTraceElement[maxDepth];
            System.arraycopy(stackTrace, 0, newTrace, 0, newTrace.length);
            ex.setStackTrace(newTrace);
        }

        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        PrintWriter writer = new PrintWriter(bos);
        ex.printStackTrace(writer);
        writer.flush();
        return bos.toString();
    }

    public void start() {
        URL parsedUrl;
        try {
            parsedUrl = new URL(serverAddress);
        } catch (Exception e) {
            throw new IllegalStateException("invalid url " + serverAddress, e);
        }
        this.host = parsedUrl.getHost();
        this.port = parsedUrl.getPort();

        // Register listener.
        if (sampleStrategy == null && radio != null && radio > 0 && radio <= 1) {
            sampleStrategy = new TraceSampleStrategy();
            sampleStrategy.setRadio(radio);
        }

        executionListener = new ManagerExecutionListener(this, sampleStrategy);
        ExecutionListenerRegistry.register(executionListener);
        UpdateContentTask.startTask(host, port);
        executorService.setMaximumPoolSize(1);
        executorService.scheduleAtFixedRate(this::export, idle, idle, TimeUnit.MILLISECONDS);
    }

    public void submit(TraceData item) {
        //noinspection ResultOfMethodCallIgnored
        traceQueue.offer(item);
    }

    public String getServerAddress() {
        return serverAddress;
    }

    public void setServerAddress(String serverAddress) {
        this.serverAddress = serverAddress;
    }

    public long getTimeout() {
        return timeout;
    }

    public void setTimeout(long timeout) {
        this.timeout = timeout;
    }

    public Map<String, String> getHeaders() {
        return headers;
    }

    public void setHeaders(Map<String, String> headers) {
        this.headers = headers;
    }

    public Double getRadio() {
        return radio;
    }

    public void setRadio(Double radio) {
        this.radio = radio;
        if (sampleStrategy != null && radio != null && radio > 0 && radio <= 1) {
            sampleStrategy.setRadio(radio);
        }
    }

    public TraceSampleStrategy getSampleStrategy() {
        return sampleStrategy;
    }

    public void setSampleStrategy(TraceSampleStrategy sampleStrategy) {
        this.sampleStrategy = sampleStrategy;
        if (this.executionListener != null) {
            this.executionListener.setSampleStrategy(this.sampleStrategy);
        }
    }
}
